package yunjiao.javatutorials.guava.concurrent.transform.assembly;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * 实时数据处理
 *
 * @author yangyunjiao
 */
public class RealTimeDataProcessor {
    private final ListeningExecutorService executor =
            MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(20));

    /**
     * 实时数据处理流水线：解析 -> 验证 -> 丰富数据 -> 存储 -> 通知
     */
    public ListenableFuture<ProcessingResult> processRealTimeDataAsync(String rawData) {
        System.out.println(">>>实时数据处理流水线：解析 -> 验证 -> 丰富数据 -> 存储 -> 通知");
        // 1. 解析数据
        ListenableFuture<ParsedData> parsedFuture = Futures.transform(
                executor.submit(() -> rawData),
                this::parseJsonData,
                executor
        );

        // 2. 验证数据格式
        ListenableFuture<ValidatedData> validatedFuture = Futures.transform(
                parsedFuture,
                this::validateData,
                executor
        );

        // 3. 丰富数据（添加时间戳、序列号等）
        ListenableFuture<EnrichedData> enrichedFuture = Futures.transform(
                validatedFuture,
                this::enrichData,
                executor
        );

        // 4. 存储到数据库（异步）
        ListenableFuture<StoredData> storedFuture = Futures.transformAsync(
                enrichedFuture,
                this::storeDataAsync,
                executor
        );

        // 5. 发送通知（如果需要）
        return Futures.transformAsync(
                storedFuture,
                this::sendNotificationIfNeededAsync,
                executor
        );
    }

    /**
     * 批量处理实时数据
     */
    public ListenableFuture<List<ProcessingResult>> processBatchDataAsync(List<String> batchData) {
        List<ListenableFuture<ProcessingResult>> futures = batchData.stream()
                .map(this::processRealTimeDataAsync)
                .collect(Collectors.toList());

        // 等待所有处理完成
        return Futures.allAsList(futures);
    }

    /**
     * 带背压控制的处理
     */
    public ListenableFuture<ProcessingResult> processWithBackpressureAsync(
            String rawData, AtomicInteger concurrentRequests) {

        // 检查并发数
        if (concurrentRequests.get() >= 100) {
            return Futures.immediateFailedFuture(
                    new RuntimeException("系统繁忙，请稍后重试")
            );
        }

        concurrentRequests.incrementAndGet();

        ListenableFuture<ProcessingResult> resultFuture = processRealTimeDataAsync(rawData);

        // 处理完成后减少计数
        return Futures.transform(
                resultFuture,
                result -> {
                    concurrentRequests.decrementAndGet();
                    return result;
                },
                MoreExecutors.directExecutor()
        );
    }

    private ParsedData parseJsonData(String rawData) {
        System.out.println("parseJsonData");
        try {
            // 模拟JSON解析
            return new ParsedData(rawData, System.currentTimeMillis());
        } catch (Exception e) {
            throw new DataParseException("数据解析失败", e);
        }
    }

    private ValidatedData validateData(ParsedData parsedData) {
        System.out.println("validateData");
        if (parsedData.timestamp() > System.currentTimeMillis()) {
            throw new ValidationException("时间戳无效");
        }
        return new ValidatedData(parsedData, true, "验证通过");
    }

    private EnrichedData enrichData(ValidatedData validatedData) {
        System.out.println("enrichData");
        return new EnrichedData(
                validatedData,
                UUID.randomUUID().toString(),
                System.currentTimeMillis(),
                "node-1"
        );
    }

    private ListenableFuture<StoredData> storeDataAsync(EnrichedData enrichedData) {
        System.out.println("storeDataAsync");
        return executor.submit(() -> {
            // 模拟数据库存储
            Thread.sleep(50);
            String recordId = "record_" + System.currentTimeMillis();
            return new StoredData(enrichedData, recordId, true);
        });
    }

    private ListenableFuture<ProcessingResult> sendNotificationIfNeededAsync(StoredData storedData) {
        System.out.println("sendNotificationIfNeededAsync");
        if (storedData.success()) {
            // 只有存储成功才发送通知
            return executor.submit(() -> {
                // 模拟发送通知
                Thread.sleep(30);
                return new ProcessingResult(
                        storedData.recordId(),
                        "SUCCESS",
                        "数据处理完成"
                );
            });
        } else {
            return Futures.immediateFuture(
                    new ProcessingResult(
                            storedData.recordId(),
                            "STORAGE_FAILED",
                            "数据存储失败"
                    )
            );
        }
    }

}
